<?php


namespace app\common\service;


/**
 * Class Zookeeper https://github.com/php-zookeeper/php-zookeeper/blob/master/examples
 * API: https://github.com/php-zookeeper/php-zookeeper/blob/v0.6.4/zookeeper-api.php
 * @package app\common\service
 *
 * @Author: Li Guo Xin <guoxinlee129@gmail.com>
 * @Date: 2020/5/14
 * @Time: 13:50
 */
class Zookeeper
{
    //protected static $zk;
    //protected static $myNode;
    //protected static $isNotifyed;

    //
    //// 获取锁
    //public static function tryGetDistributedLock($lockKey, $value)
    //{
    //    try {
    //        // 创建根节点
    //        self::createRootPath($value);
    //        // 创建临时顺序节点
    //        self::createSubPath(self::$root.$lockKey, $value);
    //        // 获取锁
    //        return self::getLock();
    //
    //    } catch (\ZookeeperException $e) {
    //        return false;
    //    } catch (\Exception $e) {
    //        return false;
    //    }
    //}
    //
    //// 释放锁
    //public static function releaseDistributedLock()
    //{
    //    if (self::$zk->delete(self::$myNode)) {
    //        return true;
    //    } else {
    //        return false;
    //    }
    //}
    //
    //public static function createRootPath($value)
    //{
    //    $aclArray = [
    //        [
    //            'perms'  => Zookeeper::PERM_ALL,
    //            'scheme' => 'world',
    //            'id'     => 'anyone',
    //        ]
    //    ];
    //    // 判断根节点是否存在
    //    if (false == self::$zk->exists(self::$root)) {
    //        // 创建根节点
    //        $result = self::$zk->create(self::$root, $value, $aclArray);
    //        if (false == $result) {
    //            throw new \Exception('create '.self::$root.' fail');
    //        }
    //    }
    //
    //    return true;
    //}
    //
    //public static function createSubPath($path, $value)
    //{
    //    // 全部权限
    //    $aclArray = [
    //        [
    //            'perms'  => Zookeeper::PERM_ALL,
    //            'scheme' => 'world',
    //            'id'     => 'anyone',
    //        ]
    //    ];
    //    /**
    //     * flags :
    //     * 0 和 null 永久节点，
    //     * Zookeeper::EPHEMERAL临时，
    //     * Zookeeper::SEQUENCE顺序，
    //     * Zookeeper::EPHEMERAL | Zookeeper::SEQUENCE 临时顺序
    //     */
    //    self::$myNode = self::$zk->create($path, $value, $aclArray, Zookeeper::EPHEMERAL | Zookeeper::SEQUENCE);
    //    if (false == self::$myNode) {
    //        throw new \Exception('create -s -e '.$path.' fail');
    //    }
    //    echo 'my node is '.self::$myNode.'-----------'.PHP_EOL;
    //
    //    return true;
    //}
    //

    ///**
    // * 通知回调处理
    // * @param $type 变化类型 Zookeeper::CREATED_EVENT, Zookeeper::DELETED_EVENT, Zookeeper::CHANGED_EVENT
    // * @param $state
    // * @param $key 监听的path
    // */
    //public static function watcher($type, $state, $key)
    //{
    //    echo PHP_EOL.$key.' notifyed ....'.PHP_EOL;
    //    self::$isNotifyed = true;
    //    self::getLock();
    //}
    //
    //public static function checkMyNodeOrBefore()
    //{
    //    $list = self::$zk->getChildren(self::$root);
    //    sort($list);
    //    $root = self::$root;
    //    array_walk($list, function (&$val) use ($root) {
    //        $val = $root.'/'.$val;
    //    });
    //
    //    if ($list[0] == self::$myNode) {
    //        echo 'get locak node '.self::$myNode.'....'.PHP_EOL;
    //        return true;
    //    } else {
    //        // 找到上一个节点
    //        $index  = array_search(self::$myNode, $list);
    //        $before = $list[$index - 1];
    //        echo 'before node '.$before.'.........'.PHP_EOL;
    //        return $before;
    //    }
    //}


    private $zookeeper;
    private static $isNotifyed;
    private $callback = array();
    public function getLock()
    {
        // 获取子节点列表从小到大，显然不可能为空，至少有一个节点
        $res = self::checkMyNodeOrBefore();
        if ($res === true) {
            return true;
        } else {
            self::$isNotifyed = false;// 初始化状态值
            // 考虑监听失败的情况：当我正要监听before之前，它被清除了，监听失败返回 false
            $result = self::$zk->get($res, [\zkCli::class, 'watcher']);
            while (!$result) {
                $res1 = self::checkMyNodeOrBefore();
                if ($res1 === true) {
                    return true;
                } else {
                    $result = self::$zk->get($res1, [\zkCli::class, 'watcher']);
                }
            }

            // 阻塞，等待watcher被执行，watcher执行完回到这里
            while (!self::$isNotifyed) {
                echo '.';
                usleep(500000); // 500ms
            }

            return true;
        }
    }

    /**
     * Zookeeper constructor.
     * @param $address CSV list of host:port values (e.g. "host1:2181,host2:2181")
     * @throws \Exception
     */
    public function __construct($address)
    {
        \Zookeeper::setDebugLevel(\Zookeeper::LOG_LEVEL_DEBUG);
        $this->zookeeper = new \Zookeeper($address);
    }

    /**
     * @param $scheme
     * @param $cert
     * @param  null  $completion_cb
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/5/14
     * @Time: 14:49
     */
    public function addAuth($scheme, $cert, $completion_cb = null)
    {
        $this->zookeeper->addAuth($scheme, $cert, $completion_cb);
    }

    /**
     * @param $path
     * @param $value
     * @param $acl
     * @param  null  $flags
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/5/14
     * @Time: 14:54
     */
    public function create($path, $value, $acl, $flags = null)
    {
        $this->zookeeper->create($path, $value, $acl, $flags);
    }

    /**
     * 设置一个节点的值，当节点不存在时，创建该节点，
     * @param $path
     * @param $value
     * @param  int  $version
     * @param  null  $stat
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/5/14
     * @Time: 14:52
     */
    public function set($path, $value, $version = -1, &$stat = null)
    {
        if (!$this->zookeeper->exists($path)) {
            $this->makePath($path);
            $this->makeNode($path, $value);
        } else {
            $this->zookeeper->set($path, $value, $version, $stat);
        }
    }

    /**
     * Equivalent of "mkdir -p" on ZooKeeper
     *
     * @param  string  $path  The path to the node
     * @param  string  $value  The value to assign to each new node along the path
     */
    public function makePath($path, $value = ''): void
    {
        $parts   = explode('/', $path);
        $parts   = array_filter($parts);
        $subPath = '';
        while (count($parts) > 1) {
            $subPath .= '/'.array_shift($parts);
            if (!$this->zookeeper->exists($subPath)) {
                $this->makeNode($subPath, $value);
            }
        }
    }

    /**
     * 在给定的目录上创建一个节点
     *
     * @param  string  $path  The path to the node
     * @param  string  $value  The value to assign to the new node
     * @param  array  $acl  Optional parameters for the Zookeeper node.
     *                       By default, a public node is created
     *
     * @return string the path to the newly created node or null on failure
     */
    public function makeNode($path, $value, array $acl = array()): string
    {
        if (empty($acl)) {
            $acl = [
                [
                    'perms'  => \Zookeeper::PERM_ALL,
                    'scheme' => 'world',
                    'id'     => 'anyone',
                ]
            ];
        }
        return $this->zookeeper->create($path, $value, $acl, $flags = null);
    }

    /**
     * 获取节点上的值
     * @param $path
     * @param  null  $watcher_cb
     * @param  null  $stat
     * @param  int  $max_size
     * @return string|null
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/5/14
     * @Time: 14:19
     */
    public function get($path, $watcher_cb = null, &$stat = null, $max_size = 0)
    {
        if (!$this->zookeeper->exists($path)) {
            return null;
        }
        return $this->zookeeper->get($path, $watcher_cb, $stat, $max_size);
    }

    /**
     * 列出给定路径的子代，即当前节点内的目录名称（如果有）
     * @param $path
     * @param  null  $watcher_cb
     * @return mixed
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/5/14
     * @Time: 14:20
     */
    public function getChildren($path, $watcher_cb = null)
    {
        if (strlen($path) > 1 && preg_match('@/$@', $path)) {
            // remove trailing /
            $path = substr($path, 0, -1);
        }
        return $this->zookeeper->getChildren($path, $watcher_cb);
    }

    /**
     * 如果没有任何子节点，删除该节点
     * @param $path
     * @param  null  $watcher_cb
     * @param  int  $version
     * @return true|null 如果节点不存在,则返回null
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/5/14
     * @Time: 14:22
     */
    public function deleteNode($path, $watcher_cb = null, $version = -1)
    {
        if (!$this->zookeeper->exists($path, $watcher_cb)) {
            return null;
        } else {
            return $this->zookeeper->delete($path, $version);
        }
    }

    /**
     * 监听给定的路径
     * @param  string  $path  the path to node
     * @param  callable  $callback  callback function
     * @return string|null
     */
    public function watch($path, $callback)
    {
        if (!is_callable($callback)) {
            return null;
        }

        if ($this->zookeeper->exists($path)) {
            if (!isset($this->callback[$path])) {
                $this->callback[$path] = array();
            }
            if (!in_array($callback, $this->callback[$path])) {
                $this->callback[$path][] = $callback;
                return $this->zookeeper->get($path, array($this, 'watchCallback'));
            }
        }
    }

    /**
     * 监听回调事件
     * @param  int  $event_type
     * @param  int  $stat
     * @param  string  $path
     * @return the return of the callback or null
     */
    public function watchCallback($event_type, $stat, $path)
    {
        if (!isset($this->callback[$path])) {
            return null;
        }

        foreach ($this->callback[$path] as $callback) {
            $this->zookeeper->get($path, array($this, 'watchCallback'));
            return call_user_func($callback);
        }
    }

    /**
     * 删除监听的回调节点，当回调不存在时，删除所有节点
     * Delete watch callback on a node, delete all callback when $callback is null
     * @param  string  $path
     * @param  callable  $callback
     * @return boolean|NULL
     */
    public function cancelWatch($path, $callback = null)
    {
        if (isset($this->callback[$path])) {
            if (empty($callback)) {
                unset($this->callback[$path]);
                $this->zookeeper->get($path); //reset the callback
                return true;
            } else {
                $key = array_search($callback, $this->callback[$path]);
                if ($key !== false) {
                    unset($this->callback[$path][$key]);
                    return true;
                } else {
                    return null;
                }
            }
        } else {
            return null;
        }
    }
}
